<?php

namespace Diablo7\HyperfPulsar;

use Diablo7\HyperfPulsar\Exception\InvalidConfigException;
use Diablo7\HyperfPulsar\Message\MessageInterface;
use Hyperf\Contract\ConfigInterface;
use Hyperf\Contract\ConnectionInterface;
use Hyperf\Contract\ContainerInterface;
use Hyperf\Pool\Pool;

class PulsarPool extends Pool
{
    /**
     * @var PulsarEndpoint
     */
    public $endpoint;

    public function __construct(ContainerInterface $container, MessageInterface $message)
    {
        parent::__construct($container, $message->getPoolConfig());

        $this->container = $container;

        $endpoint = $message->getEndpoint();
        $config = $container->get(ConfigInterface::class)->get(sprintf("pulsar.%s",$endpoint),[]);
        if(!$config){
            throw new InvalidConfigException(sprintf("缺少配置:%s",$endpoint));
        }
        $this->endpoint = new PulsarEndpoint(
           sprintf("%s/%s?%s",$config["websocket"],$message->getSocketUri(),http_build_query($message->getQueryParams())),
           $config["settings"]
        );
    }


    public function createConnection(): ConnectionInterface
    {
        return new PulsarConnection(
            $this->container,
            $this,
            new PulsarClient($this->endpoint)
        );
    }
}